package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.TradeProvinceOrderWindow;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import groupOne.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/*
drop table if exists dws_trade_province_order_window;
create table if not exists dws_trade_province_order_window
(
    `stt`           DATETIME comment '窗口起始时间',
    `edt`           DATETIME comment '窗口结束时间',
    `province_id`   VARCHAR(10) comment '省份ID',
    `province_name` VARCHAR(30) comment '省份名称',
    `cur_date`      DATE comment '当天日期',
    `order_count`   BIGINT replace comment '订单数',
    `order_amount`  DECIMAL(16, 2) replace comment '订单金额',
    `user_count`   BIGINT replace comment '订单数',
) engine = olap aggregate key (`stt`, `edt`, `province_id`, `province_name`, `cur_date`)
comment "交易域省份粒度下单各窗口汇总表"
partition by range(`cur_date`)()
distributed by hash(`province_id`) buckets 10
    properties (
  "replication_num" = "3",
  "dynamic_partition.enable" = "true",
  "dynamic_partition.time_unit" = "DAY",
  "dynamic_partition.start" = "-1",
  "dynamic_partition.end" = "1",
  "dynamic_partition.prefix" = "par",
  "dynamic_partition.buckets" = "10",
  "dynamic_partition.hot_partition_num" = "1"
);
 */
public class DwsTradeProvinceOrderWindow_huangyanhui extends BaseAppOneStream {
    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow_huangyanhui().init(
                3000,
                2,
                "DwsTradeProvinceOrderWindow_huangyanhui",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        //1.把数据封装到pojo中
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = parseToPojo(stream);
        //2.按照province 分组开窗 聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim = windowAndAgg(beanStream);

        //3.补充维度信息
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithDim = addDim(beanStreamWithoutDim);
       beanStreamWithDim.print();
        //4. 写出到doris 中
        writeToDoris(beanStreamWithDim);


    }

    private void writeToDoris(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithDim) {
        beanStreamWithDim
                .map(bean -> {
                    SerializeConfig conf = new SerializeConfig();
                    conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                    return JSON.toJSONString(bean,conf);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_trade_province_order_window"));
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> addDim(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim) {
       return AsyncDataStream.unorderedWait(
                beanStreamWithoutDim,
                new DimAsyncFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public String getTable() {
                        return "dim_base_province";
                    }

                    @Override
                    public String getId(TradeProvinceOrderWindow input) {
                        return input.getProvinceId();
                    }

                    @Override
                    public void addDim(JSONObject dim, TradeProvinceOrderWindow bean) {
                      bean.setProvinceName(dim.getString("NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
      return  beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean,ts) -> bean.getTs())
                )
                .keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow bean1,
                                                                   TradeProvinceOrderWindow bean2) throws Exception {
                                bean1.setOrderAmount(bean1.getOrderAmount().add(bean2.getOrderAmount()));
                                bean1.getOrderIdSet().addAll(bean2.getOrderIdSet());
                                bean1.getUserIdSet().addAll(bean2.getUserIdSet());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                            @Override
                            public void process(String provinceId,
                                                Context ctx,
                                                Iterable<TradeProvinceOrderWindow> elements,
                                                Collector<TradeProvinceOrderWindow> out) throws Exception {
                                TradeProvinceOrderWindow bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                bean.setOrderCount((long)bean.getOrderIdSet().size());
                                bean.setUserCount((long)bean.getUserIdSet().size());

                                out.collect(bean);

                            }
                        }
                );
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> parseToPojo(DataStreamSource<String> stream) {
        return stream
                .map(new MapFunction<String, TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow map(String value) throws Exception {
                        JSONObject obj = null;
                        try {
                            obj = JSON.parseObject(value);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        return
                                TradeProvinceOrderWindow.builder()
                                .provinceId(obj.getString("province_id"))
                                .orderIdSet(new HashSet<>(Collections.singleton(obj.getString("order_id"))))
                                .userIdSet(new HashSet<>(Collections.singleton(obj.getString("user_id"))))
                                .orderAmount(obj.getBigDecimal("split_final_amount"))
                                .ts(obj.getLong("od_ts") * 1000)
                                .build();
                    }
                });
    }
}
